package com.chapal.ib.activemq.client;

/**
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License.  You may obtain a copy of the License at
 *
 *      http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

import java.io.IOException;

import javax.jms.Connection;
import javax.jms.DeliveryMode;
import javax.jms.Destination;
import javax.jms.ExceptionListener;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageConsumer;
import javax.jms.MessageListener;
import javax.jms.MessageProducer;
import javax.jms.ObjectMessage;
import javax.jms.Session;
import javax.jms.TextMessage;
import javax.jms.Topic;

import org.apache.activemq.ActiveMQConnection;
import org.apache.activemq.ActiveMQConnectionFactory;

import com.chapal.ib.util.xml.XmlUtil;

/**
 * A simple tool for consuming messages
 * 
 * @version $Revision: 1.1.1.1 $
 */
public class ConsumerTool implements MessageListener, ExceptionListener {

	public static void main(String[] args) {
		final ConsumerTool consumerTool = new ConsumerTool();
		consumerTool.run();
	}

	private boolean running;
	private Session session;
	private Destination destination;

	private MessageProducer replyProducer;
	private boolean pauseBeforeShutdown;
	private boolean verbose = true;
	private int maxiumMessages;
	private String subject = "TOOL.DEFAULT";
	private boolean topic;
	private String user = ActiveMQConnection.DEFAULT_USER;
	private String password = ActiveMQConnection.DEFAULT_PASSWORD;
	private String url = ActiveMQConnection.DEFAULT_BROKER_URL;
	private boolean transacted;
	private boolean durable;
	private String clientId;
	private int ackMode = Session.AUTO_ACKNOWLEDGE;
	private String consumerName = "James";
	private long sleepTime;

	private long receiveTimeOut = 0;

	protected void consumeMessagesAndClose(Connection connection, Session session, MessageConsumer consumer) throws JMSException, IOException {
		System.out.println("We are about to wait until we consume: " + this.maxiumMessages + " message(s) then we will shutdown");

		for (int i = 0; (i < this.maxiumMessages) && isRunning();) {
			final Message message = consumer.receive(1000);
			if (message != null) {
				i++;
				onMessage(message);
			}
		}
		System.out.println("Closing connection");
		consumer.close();
		session.close();
		connection.close();
		if (this.pauseBeforeShutdown) {
			System.out.println("Press return to shut down");
			System.in.read();
		}
	}

	protected void consumeMessagesAndClose(Connection connection, Session session, MessageConsumer consumer, long timeout) throws JMSException, IOException {
		System.out.println("We will consume messages while they continue to be delivered within: " + timeout + " ms, and then we will shutdown");

		Message message;
		while ((message = consumer.receive(timeout)) != null)
			onMessage(message);

		System.out.println("Closing connection");
		consumer.close();
		session.close();
		connection.close();
		if (this.pauseBeforeShutdown) {
			System.out.println("Press return to shut down");
			System.in.read();
		}
	}

	synchronized boolean isRunning() {
		return this.running;
	}

	@Override
	public synchronized void onException(JMSException ex) {
		System.out.println("JMS Exception occured.  Shutting down client.");
		this.running = false;
	}

	@Override
	public void onMessage(Message message) {
		System.out.println("ConsumerTool.onMessage");
		try {
			if (message instanceof TextMessage) {
				final TextMessage txtMsg = (TextMessage) message;
				if (this.verbose) {
					String msg = txtMsg.getText();
					if (msg.length() > 50) msg = msg.substring(0, 50) + "...";
					System.out.println("Received:TextMessage:" + msg);
				}
			}
			if (message instanceof ObjectMessage) {
				final MessageObject obj = (MessageObject) ((ObjectMessage) message).getObject();

				System.out.println("Received:ObjectMessage:" + XmlUtil.toXML(obj));

			}

			if (message.getJMSReplyTo() != null) this.replyProducer.send(message.getJMSReplyTo(), this.session.createTextMessage("Reply: " + message.getJMSMessageID()));

			if (this.transacted) this.session.commit();
			else if (this.ackMode == Session.CLIENT_ACKNOWLEDGE) message.acknowledge();

		} catch (final JMSException e) {
			System.out.println("Caught: " + e);
			e.printStackTrace();
		} finally {
			if (this.sleepTime > 0) try {
				Thread.sleep(this.sleepTime);
			} catch (final InterruptedException e) {
			}
		}
	}

	public void run() {
		try {
			this.running = true;
			System.out.println("Connecting to URL: " + this.url);
			System.out.println("Consuming " + (this.topic ? "topic" : "queue") + ": " + this.subject);
			System.out.println("Using a " + (this.durable ? "durable" : "non-durable") + " subscription");

			final ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory(this.user, this.password, this.url);
			final Connection connection = connectionFactory.createConnection();
			System.out.println("Got connection");
			if (this.durable && (this.clientId != null) && (this.clientId.length() > 0) && !"null".equals(this.clientId)) connection.setClientID(this.clientId);
			System.out.println("After setting client id");
			connection.setExceptionListener(this);
			System.out.println("After setting ExceptionListener");
			connection.start();
			System.out.println("After setting ExceptionListener.connection.start");
			this.session = connection.createSession(this.transacted, this.ackMode);
			System.out.println("After setting session");
			if (this.topic) {
				System.out.println("Before creating destination.topic ");
				this.destination = this.session.createTopic(this.subject);
				System.out.println("After creating destination.topic ");
			} else {
				System.out.println("Before creating destination.queue ");
				this.destination = this.session.createQueue(this.subject);
				System.out.println("After creating destination.queue ");
			}

			this.replyProducer = this.session.createProducer(null);
			this.replyProducer.setDeliveryMode(DeliveryMode.NON_PERSISTENT);

			MessageConsumer consumer = null;
			if (this.durable && this.topic) consumer = this.session.createDurableSubscriber((Topic) this.destination, this.consumerName);
			else consumer = this.session.createConsumer(this.destination);
			System.out.println("Created Consumer");
			if (this.maxiumMessages > 0) {
				System.out.println("Maximum Messages>0");
				consumeMessagesAndClose(connection, this.session, consumer);
			} else {
				System.out.println("Maximum Messages<0");
				if (this.receiveTimeOut == 0) {
					System.out.println("receiveTimeOut == 0");
					consumer.setMessageListener(this);
				} else {
					System.out.println("receiveTimeOut != 0");
					consumeMessagesAndClose(connection, this.session, consumer, this.receiveTimeOut);
				}
			}

		} catch (final Exception e) {
			System.out.println("Caught: " + e);
			e.printStackTrace();
		}
	}

	public void setAckMode(String ackMode) {
		if ("CLIENT_ACKNOWLEDGE".equals(ackMode)) this.ackMode = Session.CLIENT_ACKNOWLEDGE;
		if ("AUTO_ACKNOWLEDGE".equals(ackMode)) this.ackMode = Session.AUTO_ACKNOWLEDGE;
		if ("DUPS_OK_ACKNOWLEDGE".equals(ackMode)) this.ackMode = Session.DUPS_OK_ACKNOWLEDGE;
		if ("SESSION_TRANSACTED".equals(ackMode)) this.ackMode = Session.SESSION_TRANSACTED;
	}

	public void setClientId(String clientID) {
		this.clientId = clientID;
	}

	public void setConsumerName(String consumerName) {
		this.consumerName = consumerName;
	}

	public void setDurable(boolean durable) {
		this.durable = durable;
	}

	public void setMaxiumMessages(int maxiumMessages) {
		this.maxiumMessages = maxiumMessages;
	}

	public void setPassword(String pwd) {
		this.password = pwd;
	}

	public void setPauseBeforeShutdown(boolean pauseBeforeShutdown) {
		this.pauseBeforeShutdown = pauseBeforeShutdown;
	}

	public void setQueue(boolean queue) {
		this.topic = !queue;
	}

	public void setReceiveTimeOut(long receiveTimeOut) {
		this.receiveTimeOut = receiveTimeOut;
	}

	public void setSleepTime(long sleepTime) {
		this.sleepTime = sleepTime;
	}

	public void setSubject(String subject) {
		this.subject = subject;
	}

	public void setTopic(boolean topic) {
		this.topic = topic;
	}

	public void setTransacted(boolean transacted) {
		this.transacted = transacted;
	}

	public void setUrl(String url) {
		this.url = url;
	}

	public void setUser(String user) {
		this.user = user;
	}

	public void setVerbose(boolean verbose) {
		this.verbose = verbose;
	}

}
